
#include <cassert>
#include <iostream>
#include "Interpreter.h"
#include <cmath>
#include <iterator>
//#include "InfixToRPN.h"

extern 
bool shunting_yard(const std::vector<std::string>& infix, std::vector<std::string>& rpn);

int Interpreter::NaturalJoin(std::vector<std::string>& tables, 
	std::vector<std::string>& whereInfix, std::vector<Tuple>& resultTuples, 
	std::string& resultName, std::vector<std::string>& whereRPN, bool& selectionRemains)
{

	std::cout << "Initiating natural join..." << std::endl;

	// get relation pointers for the tables
	std::vector<Relation*> relationPtrVec;
	for(unsigned int i = 0; i < tables.size(); ++i)
	{
		relationPtrVec.push_back(mSchemaManager.getRelation(tables[i]));
		if(relationPtrVec.back() == NULL)
		{
			std::cout << "Table pointer came out NULL for: " << tables[i] << std::endl;
			return -1;
		}
	}

	// get the sizes (in blocks and tuples) of the relations
	std::vector<std::pair<unsigned int, unsigned int> > relationBlockTupleCountVec;
	for(unsigned int i = 0; i < tables.size(); ++i)
	{
		relationBlockTupleCountVec.push_back(std::make_pair(
			relationPtrVec[i]->getNumOfBlocks(), relationPtrVec[i]->getNumOfTuples()));
	}


	//
	// get the join attrib from whereInfix
	// assuming that the natural join is specified in the WHERE clause as:
	// WHERE tbl1.attr = tbl2.attr AND ...
	// joinAttrib and joinAttrib1 allow the attribs of the two tables to have
	// different names
	//
	//std::string joinAttrib(whereInfix[2]), joinAttrib1(whereInfix[6]), resultJoinAttrib;
	std::string joinAttrib, joinAttrib1, resultJoinAttrib;
	if(tables[0].compare(whereInfix[0]) == 0)
	{
		joinAttrib = whereInfix[2];
		joinAttrib1 = whereInfix[6];
	}
	else if(tables[1].compare(whereInfix[0]) == 0)
	{
		joinAttrib = whereInfix[6];
		joinAttrib1 = whereInfix[2];
	}
	else
		assert(false);

	std::vector<std::string> sortAttrib(2);
	sortAttrib[0] = joinAttrib;
	sortAttrib[1] = joinAttrib1;

	//
	// now that we know the join attrib, we can modify the WHERE clause
	// to not have the natural join indicator sub-clause
	// erase uptil AND i.e. index 8, or uptil end of the clause, whichever is earlier
	//
	whereInfix.erase(whereInfix.begin(), 
		whereInfix.begin() + std::min(8, (int)whereInfix.size()));

	// create new schema and relation for output, then create tuples
	//std::string resultName = GetJoinedName(tables);
	resultName.assign(GetJoinedName(tables));
	std::cout << "Creating natural-joined table: " << resultName << std::endl;

	std::vector<std::string> resultFieldNames;
	std::vector<enum FIELD_TYPE> resultFieldTypes;
	GetNaturalJoinedSchema(tables, sortAttrib, 
		resultFieldNames, resultFieldTypes, resultJoinAttrib);
	Schema resultSchema(resultFieldNames, resultFieldTypes);

	std::cout << resultSchema << std::endl;
	
	if(mSchemaManager.createRelation(resultName, resultSchema) == NULL)
		return -1;
	
	Tuple resultTuple = mSchemaManager.getRelation(resultName)->createTuple();

	int totalBlocksNeeded = 0;
	for(unsigned int i = 0; i < tables.size(); ++i)
	{
		totalBlocksNeeded += relationBlockTupleCountVec[i].first;
	}

	if(totalBlocksNeeded <= mMem.getMemorySize())
	{
		////////////////////////////////////////////////
		// perform natural join in memory
		////////////////////////////////////////////////
		std::cout << "In memory operation..." << std::endl;

		// no gain by pushing the WHERE clause
		selectionRemains = true;

		// get the tuples in memory
		std::vector<std::vector<Tuple> > tuplesOfTables(tables.size());
		for(unsigned int i = 0; i < tables.size(); ++i)
		{
			relationPtrVec[i]->getBlocks(0, 0, relationBlockTupleCountVec[i].first);
			tuplesOfTables[i] = mMem.getTuples(0, relationBlockTupleCountVec[i].first);
		}

		// lets do a quadratic comparison in memory for now
		Schema schema0 = relationPtrVec[0]->getSchema();
		Schema schema1 = relationPtrVec[1]->getSchema();

		if(schema0.getFieldType(joinAttrib) != schema1.getFieldType(joinAttrib1))
		{
			std::cout << "The types of the join attributes don't match... Exiting!"
					<< std::endl;

			return -1;
		}

		bool integerJoin = false;
		if(schema0.getFieldType(joinAttrib) == INT)
			integerJoin = true;

		for(unsigned int tbl0 = 0; tbl0 < tuplesOfTables[0].size(); ++tbl0)
		{
			for(unsigned int tbl1 = 0; tbl1 < tuplesOfTables[1].size(); ++tbl1)
			{
				Tuple& tpl0 = tuplesOfTables[0][tbl0];
				Tuple& tpl1 = tuplesOfTables[1][tbl1];
			
				if(((integerJoin == true) && 
					(tpl0.getField(joinAttrib).integer == tpl1.getField(joinAttrib1).integer))
					|| ((integerJoin == false) && 
					(*(tpl0.getField(joinAttrib).str) == *(tpl1.getField(joinAttrib1).str))))

				{
					// need to join these tuples and print to screen

					if(integerJoin == true)
						resultTuple.setField(0, tpl0.getField(joinAttrib).integer);
					else
						resultTuple.setField(0, *(tpl0.getField(joinAttrib).str));

					int resultFieldIdx = 1;

					// grab the fields of the two tuples and set to result tuple
					for(int fld = 0; fld < tpl0.getNumOfFields(); ++fld)
					{
						if(schema0.getFieldName(fld).compare(joinAttrib) == 0)
							continue;

						if(resultFieldTypes[resultFieldIdx] == INT)
							resultTuple.setField(resultFieldIdx, tpl0.getField(fld).integer);
						else
							resultTuple.setField(resultFieldIdx, *(tpl0.getField(fld).str));

						++resultFieldIdx;
					}
					
					for(int fld = 0; fld < tpl1.getNumOfFields(); ++fld)
					{
						if(schema1.getFieldName(fld).compare(joinAttrib1) == 0)
							continue;

						if(resultFieldTypes[resultFieldIdx] == INT)
							resultTuple.setField(resultFieldIdx, tpl1.getField(fld).integer);
						else
							resultTuple.setField(resultFieldIdx, *(tpl1.getField(fld).str));

						++resultFieldIdx;
					}
					assert(resultFieldIdx == resultTuple.getNumOfFields());
					
					//std::cout << resultTuple << std::endl;
					resultTuples.push_back(resultTuple);
				}
			}
		}
	}
	else // not in-memory
	{
		unsigned int smallestNminus1BlockCount = 0;
		unsigned int largestBlockCount = 0, largestIdx = 0;
		for(unsigned int tbl = 0; tbl < tables.size(); ++tbl)
		{
			smallestNminus1BlockCount += relationBlockTupleCountVec[tbl].first;
			if(relationBlockTupleCountVec[tbl].first > largestBlockCount)
			{
				largestIdx = tbl;
				largestBlockCount = relationBlockTupleCountVec[tbl].first;
			}
		}
		smallestNminus1BlockCount -= largestBlockCount;

		if(smallestNminus1BlockCount <= mMem.getMemorySize() - 1)
		{
			////////////////////////////////////////////////
			// perform one pass algorithm
			////////////////////////////////////////////////
			std::cout << "One pass algorithm..." << std::endl;
			
			// no gain by pushing the WHERE clause
			selectionRemains = true;

			unsigned int smallerIdx = (largestIdx == 0) ? 1 : 0;

			// get the smaller table's tuples in memory
			relationPtrVec[smallerIdx]->getBlocks(0, 0, 
				relationBlockTupleCountVec[smallerIdx].first);

			std::vector<Tuple> smallerTableTuples = mMem.getTuples(0, 
									relationBlockTupleCountVec[smallerIdx].first);

			// lets do a quadratic comparison for now
	
			Schema schema0 = relationPtrVec[0]->getSchema();
			Schema schema1 = relationPtrVec[1]->getSchema();

			if(schema0.getFieldType(joinAttrib) != schema1.getFieldType(joinAttrib1))
			{
				std::cout << "The types of the join attributes don't match... Exiting!"
						<< std::endl;

				return -1;
			}

			bool integerJoin = false;
			if(schema0.getFieldType(joinAttrib) == INT)
				integerJoin = true;

			std::vector<Tuple> largestTableBlockTuples;

			for(unsigned int blk = 0; blk < relationBlockTupleCountVec[largestIdx].first;
						++blk)
			{
				// read one block at a time
				relationPtrVec[largestIdx]->getBlocks(blk, 0, 1);
				largestTableBlockTuples = mMem.getTuples(0, 1);

				for(unsigned int tplIdx = 0; tplIdx < largestTableBlockTuples.size(); ++tplIdx)
				{
					for(unsigned int smallerTplIdx = 0; 
						smallerTplIdx < smallerTableTuples.size(); ++smallerTplIdx)
					{
						Tuple *tpl0, *tpl1;
						if(largestIdx == 0)
						{
							tpl0 = &(largestTableBlockTuples[tplIdx]);
							tpl1 = &(smallerTableTuples[smallerTplIdx]);
						}
						else
						{
							tpl0 = &(smallerTableTuples[smallerTplIdx]);
							tpl1 = &(largestTableBlockTuples[tplIdx]);
						}
					
						if(((integerJoin == true) &&
							(tpl0->getField(joinAttrib).integer == 
									tpl1->getField(joinAttrib1).integer))

							|| ((integerJoin == false) &&
								(*(tpl0->getField(joinAttrib).str) == 
									*(tpl1->getField(joinAttrib1).str))))
						{
							// need to join these tuples and print to screen

							if(integerJoin == true)
								resultTuple.setField(0, tpl0->getField(joinAttrib).integer);
							else
								resultTuple.setField(0, *(tpl0->getField(joinAttrib).str));

							int resultFieldIdx = 1;
						
							// grab the fields of the two tuples and set to result tuple
							for(int fld = 0; fld < tpl0->getNumOfFields(); ++fld)
							{
								if(schema0.getFieldName(fld).compare(joinAttrib) == 0)
									continue;

								if(resultFieldTypes[resultFieldIdx] == INT)
									resultTuple.setField(resultFieldIdx, 
												tpl0->getField(fld).integer);
								else
									resultTuple.setField(resultFieldIdx, 
												*(tpl0->getField(fld).str));

								++resultFieldIdx;
							}
							
							for(int fld = 0; fld < tpl1->getNumOfFields(); ++fld)
							{
								if(schema1.getFieldName(fld).compare(joinAttrib1) == 0)
									continue;

								if(resultFieldTypes[resultFieldIdx] == INT)
									resultTuple.setField(resultFieldIdx, 
												tpl1->getField(fld).integer);
								else
									resultTuple.setField(resultFieldIdx, 
												*(tpl1->getField(fld).str));

								++resultFieldIdx;
							}
							assert(resultFieldIdx == resultTuple.getNumOfFields());
							
							//std::cout << resultTuple << std::endl;
							resultTuples.push_back(resultTuple);
						}
					}
				}
			}
		}
		else
		{
			////////////////////////////////////////////////
			// perform two pass algorithm
			////////////////////////////////////////////////
			std::cout << "Two pass algorithm..." << std::endl;


			//////////////////////////////////
			// Find if selection can be pushed
			//////////////////////////////////

			// start by assuming nothing can be pushed
			selectionRemains = true;

			PushClauseVecType pushClauses;
		
			unsigned int orSearchIdx = 0;
			for(; orSearchIdx < whereInfix.size(); ++orSearchIdx)
			{
				if(whereInfix[orSearchIdx].compare("or") == 0)
					break;
			}
			if(orSearchIdx == whereInfix.size())
			{
				// there is no OR, possibly can push things

				const static char* AND = "and";
				int pos, startPos = 0;
				do
				{
					for(pos = startPos; pos < whereInfix.size(); ++pos)
					{
						if(whereInfix[pos].compare(AND) == 0)
							break;
					}
					
					// startPos to pos is one sub-clause
					std::vector<unsigned int> tableNameLocs;
					for(unsigned int i = startPos; i < pos; ++i)
					{
						if(whereInfix[i].compare(".") == 0)
							tableNameLocs.push_back(i-1);
					}
					
					// determine if this sub-clause is pushable
					bool canPush = false;
					if(tableNameLocs.empty()) // when the sub-clause is stupid
						canPush = false;
					else if(tableNameLocs.size() == 1) // ideal case
						canPush = true;
					else
					{
						int tblIdx = 0;
						canPush = true; // eg. course.homework > course.project
						for(; tblIdx < tableNameLocs.size()-1; ++tblIdx)
						{
							if(whereInfix[tableNameLocs[tblIdx]].compare(
									whereInfix[tableNameLocs[tblIdx+1]]) != 0)
							{
								canPush = false;
								break;
							}
						}
					}
					
					// extract this clause if its pushable
					if(canPush)
					{
						pushClauses.push_back(std::make_pair(
							whereInfix[tableNameLocs.front()], 
							std::vector<std::string>(whereInfix.begin() + startPos, 
													whereInfix.begin() + pos)));

						// update the whereInfix by removing the pushed sub-clause
						// either remove uptil and including the "and", or uptil end
						int eatStart, eatEnd;
						if(pos == whereInfix.size())
						{
							eatEnd = pos;

							// eat the AND before it, if any
							if((startPos > 0) && (whereInfix[startPos-1].compare(AND) == 0))
								eatStart = startPos-1;
							else
								eatStart = startPos;
						}
						else
						{
							// eat the AND after it
							eatEnd = pos+1;
							eatStart = startPos;
						}
						whereInfix.erase(whereInfix.begin() + eatStart,
											whereInfix.begin() + eatEnd);

						startPos = eatStart; // stuff got eaten
					}
					else
						startPos = pos+1; // after "and"

				}
				while(startPos < whereInfix.size());
			}

			// if the whereInfix is empty, all has been pushed, selectionDuringDisplay
			// not needed
			if(whereInfix.empty() == true)
				selectionRemains = false;

			// now convert the push clauses to RPN
			PushClauseVecType pushClausesRPN(pushClauses.size());

			for(unsigned int i = 0; i < pushClauses.size(); ++i)
			{
				pushClausesRPN[i].first = pushClauses[i].first;
				if(shunting_yard(pushClauses[i].second, pushClausesRPN[i].second) == false)
				{
					std::cout << "Conversion to RPN failed!" << std::endl;
					return -1;
				}
			}

			// remove the table names and dot operators from pushClausesRPN
			for(unsigned int cl = 0; cl < pushClausesRPN.size(); ++cl)
			{
				std::vector<std::string>& clause = pushClausesRPN[cl].second;

				for(unsigned int i = 0; i < clause.size(); ++i)
				{
					if(clause[i].compare(".") == 0)
					{
						clause[i].assign(clause[i-1]);

						clause.erase(clause.begin() + i-2, clause.begin() + i);
						i -= 2; // coz the size of WHERE clause reduced by 2
					}
				}
			}


			/////////////////// DEBUG ////////////////////
			for(unsigned int i = 0; i < pushClausesRPN.size(); ++i)
			{
				std::cout << "Pushable clause for table: " << pushClausesRPN[i].first
					<< " is: " << std::endl;
				
				std::copy(pushClausesRPN[i].second.begin(), pushClausesRPN[i].second.end(),
					std::ostream_iterator<std::string, char>(std::cout, " "));
				std::cout << std::endl;
			}
			/////////////////// DEBUG ////////////////////

			///////////////////////
			// NOW RUN NATURAL JOIN
			///////////////////////

			std::vector<std::string> sortedSublists(tables.size());
			std::vector<SublistLimitsType> sublistLimitsVec(tables.size());

			CreateSortedSublists(tables[0], joinAttrib, sortedSublists[0], 
				pushClausesRPN, sublistLimitsVec[0]);

			CreateSortedSublists(tables[1], joinAttrib1, sortedSublists[1], 
				pushClausesRPN, sublistLimitsVec[1]);

			// check if the total number of sublists is prohibitive
			if(sublistLimitsVec[0].size() + sublistLimitsVec[1].size() > mMem.getMemorySize())
			{
				std::cout << "The total number of sublists is larger than M... Exiting!"
							<< std::endl;

				mSchemaManager.deleteRelation(sortedSublists[0]);
				mSchemaManager.deleteRelation(sortedSublists[1]);

				return -1;
			}

			RunNaturalJoinPass2(sortedSublists,	sublistLimitsVec, sortAttrib, 
				resultName, resultTuples);

		}
	}


	// there is still some WHERE clause
	if((selectionRemains == true) && (whereInfix.empty() == false)) 
	{
		if(shunting_yard(whereInfix, whereRPN) == true)
		{
			std::cout << "RPN notation of the WHERE clause: " << std::endl;
			for(unsigned int i = 0; i < whereRPN.size(); ++i)
			{
				std::cout << whereRPN[i] << " ";
			}
			std::cout << std::endl;
		}
		else
		{
			std::cout << "Conversion to RPN failed!" << std::endl;
			return -1;
		}

		// MODIFYING THE WHERE RPN TO RESULT SCHEMA
		// replace input table.attr by resultAttr
		for(unsigned int i = 0; i < whereRPN.size(); ++i)
		{
			if(whereRPN[i].compare(".") == 0)
			{
				if(whereRPN[i-1].compare(joinAttrib) == 0)
				{
					// course.sid becomes sid01
					whereRPN[i].assign(resultJoinAttrib);
				}
				else
				{
					// course.exam becomes exam0, course2.exam becomes exam1
					if(whereRPN[i-2].compare(tables[0]) == 0)
						whereRPN[i].assign(whereRPN[i-1].append("0"));
					else
						whereRPN[i].assign(whereRPN[i-1].append("1"));
				}
				
				whereRPN.erase(whereRPN.begin() + i-2, whereRPN.begin() + i);
				i -= 2; // coz the size of WHERE clause reduced by 2
			}
		}

		std::cout << "The modified WHERE to be applied: " << std::endl;
		std::copy(whereRPN.begin(), whereRPN.end(),
			std::ostream_iterator<std::string, char>(std::cout, " "));
		std::cout << std::endl;
	}


	std::cout << "Natural-join done! Wrapping up..." << std::endl;
	//mSchemaManager.deleteRelation(resultName);

	return 0;
}



int Interpreter::GetNaturalJoinedSchema(const std::vector<std::string>& tableNames, 
		const std::vector<std::string>& joinAttrib,
		std::vector<std::string>& resultFieldNames, 
		std::vector<enum FIELD_TYPE>& resultFieldTypes,
		std::string& resultJoinAttrib)
{

	assert(joinAttrib.size() == 2);

	char tableNumber[10];

	// add the joinAttrib as the first field to the result schema
	resultJoinAttrib.assign(joinAttrib[0]);
	resultJoinAttrib.append("01");
	resultFieldNames.push_back(resultJoinAttrib);
	resultFieldTypes.push_back(
		mSchemaManager.getRelation(tableNames[0])->getSchema().getFieldType(joinAttrib[0]));


	// now add other fields
	for(unsigned int i = 0; i < tableNames.size(); ++i)
	{
		sprintf(tableNumber, "%d", i);

		std::vector<std::string> fieldNames = 
			mSchemaManager.getRelation(tableNames[i])->getSchema().getFieldNames();

		std::vector<enum FIELD_TYPE> fieldTypes = 
			mSchemaManager.getRelation(tableNames[i])->getSchema().getFieldTypes();

		
		for(unsigned int j = 0; j < fieldNames.size(); ++j)
		{
			// ignore the join attrib
			if((fieldNames[j].compare(joinAttrib[0]) == 0) ||
				(fieldNames[j].compare(joinAttrib[1]) == 0))
				continue;

			resultFieldNames.push_back(fieldNames[j].append(tableNumber));
			resultFieldTypes.push_back(fieldTypes[j]);
		}
	}
	return 0;
}

int Interpreter::CreateSortedSublists(const std::string& table, const std::string& sortAttrib,
							std::string& sortedTable, PushClauseVecType& pushClausesRPN,
							SublistLimitsType& sublistLimits)
{
	std::vector<std::vector<std::string> > pushClausesForMe;
	for(unsigned int i = 0; i < pushClausesRPN.size(); ++i)
	{
		if(pushClausesRPN[i].first.compare(table) == 0)
			pushClausesForMe.push_back(pushClausesRPN[i].second);
	}

	sortedTable.assign(table);
	sortedTable.append("_sortedSublists");

	Relation* relationPtr = mSchemaManager.getRelation(table);

	// create the first pass output relation with same schema as the input table
	if(mSchemaManager.createRelation(sortedTable, relationPtr->getSchema()) == NULL)
		return -1;

	Relation* sortedRelationPtr = mSchemaManager.getRelation(sortedTable);


	const int numBlocks = relationPtr->getNumOfBlocks();
	int blockCtr = 0;
	std::vector<Tuple> tuplesOfSublist;
	
	// Determine whether the sortAttrib is INT or STR20
	bool intSort = false;
	if(relationPtr->getSchema().getFieldType(sortAttrib) == INT)
		intSort = true;
	
	// now create the Sorter
	SortFunctor sorter(sortAttrib, intSort);

	while(blockCtr < numBlocks)
	{
		int blocksToCopy = std::min(numBlocks - blockCtr, mMem.getMemorySize());
		relationPtr->getBlocks(blockCtr, 0, blocksToCopy);
		blockCtr += blocksToCopy;

		tuplesOfSublist = mMem.getTuples(0, blocksToCopy);

		if(pushClausesForMe.empty() == false)
		{
			std::vector<Tuple> filteredTuples;
			for(unsigned int i = 0; i < tuplesOfSublist.size(); ++i)
			{
				unsigned int j = 0;
				for(; j < pushClausesForMe.size(); ++j)
				{
					if(CheckWhereOnTuple(tuplesOfSublist[i], pushClausesForMe[j]) == false)
						break;
				}
				if(j == pushClausesForMe.size())
					filteredTuples.push_back(tuplesOfSublist[i]);
			}
			std::cout << "Filtering tuple count from: " << tuplesOfSublist.size()
						<< " to: " << filteredTuples.size() << std::endl;

			std::sort(filteredTuples.begin(), filteredTuples.end(), sorter);
			mMem.setTuples(0, filteredTuples);

			// write those memory blocks to disk
			int tuples_per_block=filteredTuples.front().getTuplesPerBlock();
			int num_blocks=filteredTuples.size()/tuples_per_block;
			int num_additional_blocks=(filteredTuples.size()%tuples_per_block>0?1:0);

			// the last one might not be full here FYI
			blocksToCopy = num_blocks + num_additional_blocks; 

			// save the blockStart and blockEnd for this sublist, to be used by the Pass2()
			sublistLimits.push_back(std::make_pair(
									sortedRelationPtr->getNumOfBlocks(), 
									sortedRelationPtr->getNumOfBlocks() + blocksToCopy));

			sortedRelationPtr->setBlocks(sortedRelationPtr->getNumOfBlocks(), 0, blocksToCopy);
		}
		else
		{
			std::sort(tuplesOfSublist.begin(), tuplesOfSublist.end(), sorter);

			mMem.setTuples(0, tuplesOfSublist);

			// save the blockStart and blockEnd for this sublist, to be used by the Pass2()
			sublistLimits.push_back(std::make_pair(
									sortedRelationPtr->getNumOfBlocks(), 
									sortedRelationPtr->getNumOfBlocks() + blocksToCopy));

			// Since there are no holes and no pushed selection, the number of blocks to 
			// write after sorting will be equal to the number of blocks read
			sortedRelationPtr->setBlocks(sortedRelationPtr->getNumOfBlocks(), 0, blocksToCopy);
		}
	}

#if 0	///// DEBUG /////

	std::cout << "The relation " << table << " contains:\n" << *relationPtr << std::endl;
	std::cout << "The sortedSublistRelation " << sortedTable << " contains:\n" 
		<< *sortedRelationPtr << std::endl;

#endif ///// DEBUG /////

	return 0;
}

int Interpreter::RunNaturalJoinPass2(const std::vector<std::string>& sortedSublists,
	const std::vector<SublistLimitsType>& sublistLimitsVec,
	const std::vector<std::string>& sortAttrib, const std::string& resultTable,
	std::vector<Tuple>& resultTuples)
{

/*	 1) Find # of sublists of each table
	 2) Allocate sublist vector for each
	 3) Initialize each sublist by reading sublist block and filling queue
	 4) Pick smallest tuple of attrib attr from head of "all" sublists.
	 	If unable to pick a smallest tuple, all queues are empty, quit
	 5) Populate vec0 with all tuples from tbl0 with attrib attr,
	 	vec1 with all tuples from tbl1 with attrib attr
		Remove those heads from sublists of tbl0 and tbl1 while populating
		If a queue becomes empty, refill it by reading from the disk if possible
	 6) If vec0 and vec1 are not empty, JOIN(vec0, vec1, resultVec)
	 	Then go to step (4)
*/

	//////////////////////////
	// INITIALIZE
	//////////////////////////

	Relation* relationPtr[2];
	std::vector<Sublist> sublistVec[2];

	for(unsigned int tbl = 0; tbl < 2; ++tbl)
	{
		relationPtr[tbl] = mSchemaManager.getRelation(sortedSublists[tbl]);

		sublistVec[tbl].resize(sublistLimitsVec[tbl].size());

		for(unsigned int sub = 0; sub < sublistVec[tbl].size(); ++sub)
		{
			Sublist& sublist = sublistVec[tbl][sub];

			sublist.blockIdx = sublistLimitsVec[tbl][sub].first;
			sublist.blockEnd = sublistLimitsVec[tbl][sub].second;

			sublist.memBlockIdx = tbl*sublistVec[0].size() + sub; // 0,1,2 and 3,4,5

			relationPtr[tbl]->getBlock(sublist.blockIdx, sublist.memBlockIdx);
			std::vector<Tuple> tempTuples = mMem.getTuples(sublist.memBlockIdx, 1);
			for(unsigned int tpl = 0; tpl < tempTuples.size(); ++tpl)
			{
				sublist.tuples.push(tempTuples[tpl]);
			}
		}
	}

	//////////////////////////
	// RUN ALGORITHM
	//////////////////////////

	Schema schema0 = relationPtr[0]->getSchema();
	Schema schema1 = relationPtr[1]->getSchema();

	if(schema0.getFieldType(sortAttrib[0]) != schema1.getFieldType(sortAttrib[1]))
	{
		std::cout << "The types of the join attributes don't match... Exiting!"
				<< std::endl;

		return -1;
	}

	bool integerJoin = false;
	if(schema0.getFieldType(sortAttrib[0]) == INT)
		integerJoin = true;

	int smallestAttribInt;
	const int MAX_INT = std::numeric_limits<int>::max();
	std::string smallestAttribStr;
	const std::string MAX_STR("zzzzzzzzzz");

	Tuple resultTuple = mSchemaManager.getRelation(resultTable)->createTuple();
	std::vector<enum FIELD_TYPE> resultFieldTypes = resultTuple.getSchema().getFieldTypes();
	int resultFieldIdx;


	while(1)
	{
		smallestAttribInt = MAX_INT;
		smallestAttribStr = MAX_STR;

		// find the smallest sortAttrib value from the heads of sublists
		for(unsigned int tbl = 0; tbl < 2; ++tbl)
		{
			for(unsigned int sub = 0; sub < sublistVec[tbl].size(); ++sub)
			{
				Sublist& sublist = sublistVec[tbl][sub];

				if(sublist.tuples.empty() == false)
				{
					if(integerJoin == true)
					{
						if(sublist.tuples.front().getField(sortAttrib[tbl]).integer
								< smallestAttribInt)
						{
							smallestAttribInt = 
								sublist.tuples.front().getField(sortAttrib[tbl]).integer;
						}
					}
					else
					{
						if(*(sublist.tuples.front().getField(sortAttrib[tbl]).str)
								< smallestAttribStr)
						{
							smallestAttribStr = 
								*(sublist.tuples.front().getField(sortAttrib[tbl]).str);
						}

					}
				}
			}
		}

		// check if all queues are empty, so we can exit this massive while(1)
		if(integerJoin == true)
		{
			if(smallestAttribInt == MAX_INT)
				break;
		}
		else
		{
			if(smallestAttribStr == MAX_STR)
				break;
		}
	
		// extract all tuples that have the smallest value for the sortAttrib
		// while deleting from the queue
		std::vector<Tuple> tuplesToJoin[2];
		for(unsigned int tbl = 0; tbl < 2; ++tbl)
		{
			for(unsigned int sub = 0; sub < sublistVec[tbl].size(); ++sub)
			{
				Sublist& sublist = sublistVec[tbl][sub];
				
				while((sublist.tuples.empty() == false) &&
					 (((integerJoin == true) && 
					 	(sublist.tuples.front().getField(sortAttrib[tbl]).integer
															== smallestAttribInt))
					 || ((integerJoin == false) &&
					 	(*(sublist.tuples.front().getField(sortAttrib[tbl]).str)
															== smallestAttribStr))))
				{
					//
					// the tuples with smallestAttrib value could be across the 
					// block boundary, hence if popping makes a block empty, it
					// needs to be read and queue refilled right here, if its possible
					// to do so
					//
					tuplesToJoin[tbl].push_back(sublist.tuples.front());
					sublist.tuples.pop();

					if(sublist.tuples.empty())
					{
						++sublist.blockIdx;
						if(sublist.blockIdx < sublist.blockEnd)
						{
							relationPtr[tbl]->getBlock(sublist.blockIdx, sublist.memBlockIdx);
							std::vector<Tuple> tempTuples = 
										mMem.getTuples(sublist.memBlockIdx, 1);
							for(unsigned int tpl = 0; tpl < tempTuples.size(); ++tpl)
							{
								sublist.tuples.push(tempTuples[tpl]);
							}
						}
					}
				}
			}
		}
	
		if((tuplesToJoin[0].empty() == false) && (tuplesToJoin[1].empty() == false))
		{
			//////////////////
			// JOIN THE TUPLES
			//////////////////

			if(integerJoin == true)
				resultTuple.setField(0, smallestAttribInt); // common field 0 for all
			else
				resultTuple.setField(0, smallestAttribStr); // common field 0 for all

			for(unsigned int i = 0; i < tuplesToJoin[0].size(); ++i)
			{
				resultFieldIdx = 1; // start from field 1
			
				Tuple& tpl0 = tuplesToJoin[0][i];
				
				// grab the fields of the first tuple and set to result tuple
				for(int fld = 0; fld < tpl0.getNumOfFields(); ++fld)
				{
					if(schema0.getFieldName(fld).compare(sortAttrib[0]) == 0)
						continue;

					if(resultFieldTypes[resultFieldIdx] == INT)
						resultTuple.setField(resultFieldIdx, 
									tpl0.getField(fld).integer);
					else
						resultTuple.setField(resultFieldIdx, 
									*(tpl0.getField(fld).str));

					++resultFieldIdx;
				}
				assert(resultFieldIdx == tpl0.getNumOfFields());


				for(unsigned int j = 0; j < tuplesToJoin[1].size(); ++j)
				{
					resultFieldIdx = tpl0.getNumOfFields(); // start from after tpl0 fields

					Tuple& tpl1 = tuplesToJoin[1][j];
					
					// grab the fields of the second tuple and set to result tuple
					for(int fld = 0; fld < tpl1.getNumOfFields(); ++fld)
					{
						if(schema1.getFieldName(fld).compare(sortAttrib[1]) == 0)
							continue;

						if(resultFieldTypes[resultFieldIdx] == INT)
							resultTuple.setField(resultFieldIdx, 
										tpl1.getField(fld).integer);
						else
							resultTuple.setField(resultFieldIdx, 
										*(tpl1.getField(fld).str));

						++resultFieldIdx;
					}
					assert(resultFieldIdx == resultTuple.getNumOfFields());
					
					//std::cout << resultTuple << std::endl;
					resultTuples.push_back(resultTuple);
				}
			}
		} // end if

	} // end while

	// the sorted sublists aren't needed any more
	mSchemaManager.deleteRelation(sortedSublists[0]);
	mSchemaManager.deleteRelation(sortedSublists[1]);

	return 0;
}







